package com.booter.producer.config.sentinel;

import com.alibaba.csp.sentinel.datasource.AbstractDataSource;
import com.alibaba.csp.sentinel.datasource.Converter;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
public class CustomerRedisDataSource extends AbstractDataSource<String, List<FlowRule>> {

    private RedisMessageListenerContainer container;
    private ObjectMapper objectMapper = new ObjectMapper();
    private Map<String, FlowRule> rulesMap;
    private RedisTemplate<String, String> redisTemplate;
    private String key;
    private Converter<String, List<FlowRule>> converter;

    public CustomerRedisDataSource(RedisTemplate<String, String> redisTemplate, String key, String channel, Converter<String, List<FlowRule>> converter, RedisMessageListenerContainer container) {
        super(converter);
        this.converter = converter;
        this.container = container;
        this.redisTemplate = redisTemplate;
        this.key = key;
        this.rulesMap = new ConcurrentHashMap<>();
        this.objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        this.container.addMessageListener(new MessageListenerAdapter((MessageListener) (message, pattern) -> {
            byte[] body = message.getBody();
            try {
                List<FlowRule> list = this.converter.convert(new String(body));
                list.stream().filter(s -> {
                    if (!this.rulesMap.containsKey(s.getResource())) {
                        return true;
                    } else {
                        this.rulesMap.put(s.getResource(), s);
                        return false;
                    }
                }).forEach(s -> {
                    this.rulesMap.put(s.getResource(), s);
                });
                list = this.loadConfig();
                this.redisTemplate.opsForValue().set(this.key, this.objectMapper.writeValueAsString(list));
                this.getProperty().updateValue(list);
            } catch (Exception ex) {
                log.error("加载增量规则异常", ex);
            }

        }), new PatternTopic(channel));
        this.loadInitialConfig();
    }

    private void loadInitialConfig() {
        this.loadConfig();
    }


    @Override
    public List<FlowRule> loadConfig() {
        if (!this.rulesMap.isEmpty()) {
            return this.rulesMap.entrySet().stream().map(s -> s.getValue()).collect(Collectors.toList());
        }
        String s = this.readSource();
        if (StringUtils.isBlank(s)) {
            return new ArrayList<>();
        }
        List<FlowRule> list = this.converter.convert(s);
        if (!CollectionUtils.isEmpty(list)) {
            this.rulesMap.putAll(list.stream().collect(Collectors.toConcurrentMap(FlowRule::getResource, Function.identity())));
        }
        this.getProperty().updateValue(list);
        return list;
    }

    @Override
    public String readSource() {
        return this.redisTemplate.opsForValue().get(this.key);
    }

    @Override
    public void close() {
        //do nothing
    }
}
